package com.amin.zookeeper;

import java.util.List;
import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.AsyncCallback.StringCallback;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import org.junit.Test;

public class ZkApiDemo {
	static {

	}
	public static String hostAndPort = "192.168.186.3:2181";

	/**
	 * 创建连接 connectString连接zk服务器的ip 192.168.186.3:2181
	 * sessionTimeout:回话的超时时间，单位为毫秒 watcher:监听器对象
	 * 
	 * @throws Exception
	 */
	@Test
	public void testConect() throws Exception {
		final CountDownLatch cdl = new CountDownLatch(1);
		ZooKeeper zk = new ZooKeeper(hostAndPort, 3000, new Watcher() {
			@Override
			public void process(WatchedEvent event) {
				System.out.println("连接成功");
				cdl.countDown();
			}
		});
		// while(true);
		cdl.await();
	}

	@Test
	public void testCreate() throws Exception {
		final CountDownLatch cdl = new CountDownLatch(1);
		ZooKeeper zk = new ZooKeeper(hostAndPort, 3000, new Watcher() {
			@Override
			public void process(WatchedEvent event) {
				System.out.println("连接成功");
				cdl.countDown();
			}
		});
		cdl.await();
		// path：节点路径
		// data:节点的内容
		// acl：节点的权限
		// createMode：节点的类型
		// PERSISTENT(persistent)：普通持久节点
		// EPHEMERAL(ephemeral):普通临时节点
		// PERSISTENT_SEQUENTIAL:顺序持久节点
		// EPHEMERAL_SEQUENTIAL:顺序临时节点
		zk.create("/zk01", "hellozk".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
		System.out.println("创建节点成功");
		while (true)
			;
	}

	@Test
	public void testGet() throws Exception {
		final CountDownLatch cdl = new CountDownLatch(1);
		ZooKeeper zk = new ZooKeeper(hostAndPort, 3000, new Watcher() {
			@Override
			public void process(WatchedEvent event) {
				System.out.println("连接成功");
				cdl.countDown();
			}
		});
		cdl.await();
		// path:获取节点的地址
		// watcher监听器
		// stat:查询后，会将节点的其它信息封装stat中
		Stat stat = new Stat();
		byte[] data = zk.getData("/zk01", new Watcher() {
			@Override
			public void process(WatchedEvent event) {
				System.out.println("获取到数据");
			}
		}, stat);
		System.out.println(new String(data));
		System.out.println(stat);
	}

	@Test
	// 修改节点
	public void testSet() throws Exception {
		final CountDownLatch cdl = new CountDownLatch(1);
		ZooKeeper zk = new ZooKeeper(hostAndPort, 3000, new Watcher() {
			@Override
			public void process(WatchedEvent event) {
				System.out.println("连接成功");
				cdl.countDown();
			}
		});
		cdl.await();
		// data:修改后节点的内容。
		// version:修改节点对应的版本号
		zk.setData("/zk01", "you can't sleep".getBytes(), -1);
		while (true)
			;
	}

	@Test
	public void testCreateChildNode() {
		try {
			final CountDownLatch cdl = new CountDownLatch(1);
			ZooKeeper zk = new ZooKeeper(hostAndPort, 3000, new Watcher() {
				@Override
				public void process(WatchedEvent event) {
					System.out.println("连接成功");
					cdl.countDown();
				}
			});
			cdl.await();
			zk.create("/zk01/zk001", "I am Yuqian".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
			while (true)
				;
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	@Test
	public void testGetChild() {
		try {

			final CountDownLatch cdl = new CountDownLatch(1);
			ZooKeeper zk = new ZooKeeper(hostAndPort, 3000, new Watcher() {
				@Override
				public void process(WatchedEvent event) {
					System.out.println("连接成功");
					cdl.countDown();
				}
			});
			cdl.await();
			// 注意：stat是zk01的节点信息
			Stat stat = new Stat();
			List<String> paths = zk.getChildren("/zk01", null, stat);
			for (String path : paths) {
				System.out.println(path);
			}
			System.out.println(stat);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	@Test
	public void testDelete() throws Exception {
		final CountDownLatch cdl = new CountDownLatch(1);
		ZooKeeper zk = new ZooKeeper(hostAndPort, 3000, new Watcher() {
			@Override
			public void process(WatchedEvent event) {
				System.out.println("连接成功");
				cdl.countDown();
			}
		});
		cdl.await();
		/**
		 * path:删除的节点路径 version:节点的版本，-1支持各种版本号 删除节点使用的场景不是太多，可以创建临时节点，连接断了之后，
		 * 节点会自动被删除。 注意：只用空节点才能删除。
		 */
		zk.delete("/zk01/zk001", -1);
	}
	
	/**
	 * 监听节点发生变化
	 * @throws Exception
	 */
	@Test
	public void testGetDataWatcher() throws Exception {
		final CountDownLatch cdl = new CountDownLatch(1);
		ZooKeeper zk = new ZooKeeper(hostAndPort, 3000, new Watcher() {
			@Override
			public void process(WatchedEvent event) {
				System.out.println("连接成功");
				cdl.countDown();
			}
		});
		cdl.await();
		
		while (true) {
			final CountDownLatch cdl2 = new CountDownLatch(1);
			zk.getData("/zk01", new Watcher() {
				@Override
				public void process(WatchedEvent event) {
					if(event.getType()==EventType.NodeDataChanged){
		                  System.out.println("节点发生了变化");
		              }
					System.out.println("数据放生变化");
					cdl2.countDown();
				}
			}, null);
			cdl2.await();
		}
	}
	
	/**
	 * 创建回调
	 * @throws Exception
	 */
	@Test
	public void testCreateCallBack() throws Exception {
		final CountDownLatch cdl = new CountDownLatch(1);
		ZooKeeper zk = new ZooKeeper(hostAndPort, 3000, new Watcher() {
			@Override
			public void process(WatchedEvent event) {
				System.out.println("连接成功");
				cdl.countDown();
			}
		});
		cdl.await();
		/**
		 * path:节点路径 data:节点内容 acl:节点的相关权限 createMode:节点的类型 ctx:传入的附件
		 */
		zk.create("/zk02", "Kill sleeper".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT,
				new StringCallback() {
					@Override
					public void processResult(int rc, String path, Object ctx, String name) {
						// int rc:状态码 0代表创建成功，-110 代表创建失败
						// String path：用户指定的路径
						// Object ctx：传入的附件
						// String name：实际创建节点的路径名
						System.out.println("rc:" + rc + ",path:" + path + ",ctx:" + ctx + ",name:" + name);
					}

				}, "over");
		while (true)
			;
	}
	
	
	
}
